package com.atguigu.flink.chapter11;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.OverWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.*;

public class Flink15_Window_Over_SQL {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
               env.setParallelism(1);

               //因为加了window 后变成普通的流的，不是source？？？！！！ 不是 DataStreamSource 了
           DataStream<WaterSensor> waterSensorStream =
                   env.fromElements(
                           new WaterSensor("sensor_1", 1000L, 10),
                           new WaterSensor("sensor_1", 5000L, 20),
                           new WaterSensor("sensor_1", 4000L, 40),
                           new WaterSensor("sensor_1", 4000L, 50),
                           new WaterSensor("sensor_2", 6000L, 30),
                           new WaterSensor("sensor_2", 7000L, 60))
                   .assignTimestampsAndWatermarks(
                           WatermarkStrategy
                                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                   //事件事件来自于时间戳
                                   .withTimestampAssigner((event, timestamp) -> event.getTs())
                   );


           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //  每隔5s 统计最近5s 内 每个传感器的水位和
           Table table = tableEnv.fromDataStream(waterSensorStream, $("id"), $("ts").rowtime(), $("vc"));
           // sum(vc ) over(partition by id order by ts rows between unbounded preceding  and current row)

           tableEnv.createTemporaryView("sensor",table);
         tableEnv.sqlQuery("select " +
                 " id ," +
                 "  ts," +
                 " vc ," +
             //    " sum(vc ) over( partition by id order by ts rows between unbounded preceding and current row) sum_vc" +

                 // 1 是行数，如果是时间就需要加interval
                 //  " sum(vc ) over( partition by id order by ts rows between 1 preceding and current row) sum_vc" +

                 // 根据range (时间) 来
                 // " sum(vc ) over( partition by id order by ts range between unbounded preceding and current row) sum_vc" +
                 // 往前 1s  同时求最大值vc
                  " sum(vc ) over( partition by id order by ts range between interval '1' second preceding and current row) sum_vc," +
                  // 在同一sql 语句中，所有的聚合必须是同一窗口
                 " max(vc ) over( partition by id order by ts range between interval '1' second preceding and current row) max_vc," +
                 " min(vc ) over( partition by id order by ts range between interval '1' second preceding and current row) min_vc" +
                 " from sensor")
                 // 后面可以单独创建一个窗口，类似于一个变量
                 // window w as (partition by id order by ts range between interval '1' second preceding and current row)  之后直接调用w 就行
                 .execute()
                 .print();




       }
}
